/*
 * SocketAcceptor.java    0.0.1    22/set/2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */ 
package org.mentalsmash.tazio.net.cs.server.impl;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.SocketAddress;
import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashSet;
import java.util.Set;

import org.mentalsmash.tazio.net.cs.CSSubsystemException;
import org.mentalsmash.tazio.net.cs.RuntimeCSSubsystemException;
import org.mentalsmash.tazio.net.cs.server.ProtocolHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * The SocketAcceptor is an implementation of {@link Acceptor Acceptor} that runs in a thread
 * of its own.
 * <p>
 * A <code>SocketAcceptor</code> is associated on a socket on which to listen for incoming connections.
 * The thread that runs the acceptor is started calling the {@link SocketAcceptor#start() start()}
 * method on it: it creates a thread that listens for connections on the given socket and dispatches new
 * connection to the dispatcher associated to the acceptor.
 * </p>
 * <p>
 * The association Acceptor-Dispatcher <strong>has to be</strong> set before the SocketAcceptor thread is
 * started, and cannot be changed while it is running.
 * </p>
 * @version    0.0.1    22/set/2009    
 * @author     Andrea Reale <andrea.reale@gmail.com>
 */
public class SocketAcceptor implements Acceptor, Runnable {

    private final static Logger log = LoggerFactory.getLogger(SocketAcceptor.class);

    private final static String THREAD_NAME = "Acceptor";

    private final Set<AcceptorListener> registeredListeners; //the listeners registered on the acceptors.
    //Accesses to this set is synchronised using the set's lock.

    private volatile boolean running = false; // controls the listening cycle

    private Dispatcher dispatcher; // where to dispatch incoming connections
    private ServerSocketChannel incomingSocket; // the socket to listen for connections on

    private Class<? extends ProtocolHandler> protocolHandlerClass; // The class that will be used to
    // instantiate protocol handler objects

    private Thread acceptorThread = null; // keeps a reference to the acceptor thread, if it is running;

    private final Object statusLock = new Object(); //Used to control accesses on private dispatcher
    //and protocol handler class


    private SocketAddress address; //used to trye to restore the sercer socket upon exceptions

    /**
     * Create a new <code>SocketAcceptor</code> and sets the specified values.
     * <p>
     * If the  <code>dispatcher</code> needs to be started before dispatching, you <strong>must</strong>
     * call its <code>start()</code> method prior to calling this constructor. <strong>Notice</strong> that
     * if the dispatcher is not in the <code>dispatching</code> no error is thrown, and only a warning
     * message is logged.
     * </p>
     * <p>
     * The passed <code>socketChan</code> needs to be open and its associated socket already bound.
     * </p>
     * <p>
     * The <code>handlerClass</code> must be a ProtocolHandler implementation and must provide a 
     * default constructor.
     * </p>
     * @param dispatcher the dispatcher that will be used by this acceptor to handle new connections. 
     *                   It <strong>must</strong> already have been started
     * @param handlerClass the class of the protocol handlers that are to be created to handle messages
     *                     coming from connections. It must have a default constructor.
     * @param socketChan a {@link java.nio.channels.ServerSocketChannel ServerSocketChannel} to listen on.
     *               The channel <strong>must</strong> be open, and its associated socket <strong>must</strong>
     *               be bound to a valid address and port.
     * @throws IOException if the <code>socket</code> is not bound.
     * @throws CSSubsystemException If the dispatcher is not started.
     * 
     */
    public SocketAcceptor(Dispatcher dispatcher, Class<? extends ProtocolHandler> handlerClass,
	    ServerSocketChannel socketChan) throws IOException, CSSubsystemException {

	if(!socketChan.isOpen()){
	    String msg = "SocketChannel passed as argument must be open";
	    log.error(msg);
	    throw new IOException(msg);
	}

	if(!socketChan.socket().isBound()) {
	    log.error("The socket associated to the passed socket channel must be bound");
	    throw new IOException("The socket" + socketChan +" was not bound");
	}


	registeredListeners = new HashSet<AcceptorListener>();

	setDispatcher(dispatcher);

	protocolHandlerClass = handlerClass;

	incomingSocket = socketChan;
	address = socketChan.socket().getLocalSocketAddress();
    }


    /**
     * Create a new <code>SocketAcceptor</code> and sets the specified values.
     * <p>
     * If the  <code>dispatcher</code> needs to be started before dispatching, you <strong>must</strong>
     * call its <code>start()</code> method prior to calling this constructor.<strong>Notice</strong> that
     * if the dispatcher is not in the <code>dispatching</code> no error is thrown, and only a warning
     * message is logged.
     * </p>
     * <p>
     * The <code>handlerClass</code> must be a ProtocolHandler implementation and must provide a 
     * default constructor.
     * </p>
     * @param dispatcher the dispatcher that will be used by this acceptor to handle new connections. 
     *                   It <strong>must</strong> already have been started.
     * @param handlerClass the class of the protocol handlers that are to be created to handle messages
     *                     coming from connections. It must have a default constructor.
     * @param address a valid InetAddres: it will be used to bind a socket on.
     * @param port a port for the socket to listen on
     * @throws IOException if a socket on the given <code>address</code> and <code>port</code> cannot be bound.
     * @throws CSSubsystemException If the dispatcher is not started.
     */
    public SocketAcceptor(Dispatcher dispatcher, Class<? extends ProtocolHandler> handlerClass,
	    InetAddress address, int port) throws IOException,CSSubsystemException {


	registeredListeners = new HashSet<AcceptorListener>();


	setDispatcher(dispatcher);

	incomingSocket = ServerSocketChannel.open();
	ServerSocket socket = incomingSocket.socket();

	this.address = new InetSocketAddress(address, port);
	socket.bind(this.address);

	protocolHandlerClass = handlerClass;
    }


    /**
     * Sets the class that will be used to instantiate {@link ProtocolHandler ProtocolHandler} objects, which in turn are used
     * to handle incoming data from client connections.
     * @param protocolHandlerClass A class instance which implements {@link ProtocolHandler ProtocolHandler}
     * 							   . It has to provide a default constructor.
     */
    //CONCURRENCY: 	any thread could call this. Modifications to the protocol handler, must be mutually exclusive
    //				wrt creation of new ProtocolHandler instances. 
    public void setProtocolHandlerClass(Class<? extends ProtocolHandler> protocolHandlerClass) {
	synchronized (statusLock) {
	    this.protocolHandlerClass = protocolHandlerClass;
	}
    }

    /* (non-Javadoc)
     * @see org.mentalsmash.tazio.net.cs.server.Acceptor#getCurrentDispatcher()
     */
    //CONCURRENCY:	any thread could call this. Accesses and Modifications to the current dispatcher
    //				have to be synchronized wrt registrations on the dispatcher
    @Override
    public Dispatcher getCurrentDispatcher() {
	synchronized (statusLock) {
	    return dispatcher;
	}
    }

    /* (non-Javadoc)
     * @see org.mentalsmash.tazio.net.cs.server.Acceptor#setDispatcher(org.mentalsmash.tazio.net.cs.server.Dispatcher)
     */
    @Override
    //CONCURRENCY:	any thread could call this. Accesses and Modifications to the current dispatcher
    //				have to be synchronized wrt registrations on the dispatcher
    public void setDispatcher(Dispatcher dispatcher) throws CSSubsystemException {

	if(isRunning()) {
	    String msg = "Cannot set the dispatcher for this acceptor: the acceptor is already running";
	    log.error(msg);
	    throw new CSSubsystemException(msg);
	}

	synchronized (statusLock) {
	    //FIXME: maybe I should have not commented this!!
	    //			if(!dispatcher.isDispatching()) {
	    //				throw new CSSubsystemException(new DispatcherNotRunningException());
	    //			}
	    this.dispatcher = dispatcher;
	}

    }





    /* (non-Javadoc)
     * @see org.mentalsmash.tazio.net.cs.server.Acceptor#isRunning()
     */
    //CONCURRENCY: running is volatile :)
    @Override
    public boolean isRunning() {
	return running;
    }


    /* (non-Javadoc)
     * @see org.mentalsmash.tazio.net.cs.server.Acceptor#start()
     */
    //CONCURRENCY: 	since only a thread at once can run this acceptor, this method is synchronised
    // 				on the acceptor itself to avoid multiple threads start a new acceptor
    @Override
    public Thread start() {
	synchronized (this) {


	    if(running) {
		throw new RuntimeCSSubsystemException("One Acceptor is already running");
	    }

	    acceptorThread = new Thread(this);
	    //acceptorThread.setName(THREAD_NAME);

	    acceptorThread.start();
	    while(!running) {
		try {
		    log.debug("Waiting for acceptor thread to start");
		    wait(); //waiting for the acceptorThread to actually start
		} catch (InterruptedException e) {
		    log.error("Unexpected Interrupt received ",e);
		    continue;
		}
	    }
	}
	//notify the listeners
	notifyStart();

	return acceptorThread;
    }


    /**
     * Notifies the registered listeners that the acceptor has started.
     * This method is called by the thread that made the Dispatcher Thread start
     */
    private void notifyStart() {
	synchronized (registeredListeners) {
	    for(AcceptorListener l : registeredListeners) {
		l.acceptorStarted();
	    }
	}
    }

    /* (non-Javadoc)
     * @see org.mentalsmash.tazio.net.cs.server.Acceptor#stop()
     */
    //CONCURRENCY:	synchronised on the acceptor itself: not sure if that it is strictly necessary
    //				to avoid multiple threads call stop on the thread. But it won't be harmful.
    @Override
    public synchronized void stop() {
	if(!running){
	    log.debug("Cannot sto acceptor: it is not running.");
	    return;
	}

	log.debug("Acceptor was asked to stop, working for it...");

	running = false;

	if(acceptorThread != null && acceptorThread.isAlive()) {
	    acceptorThread.interrupt(); // if the serversocket is blocked on a accept() raises the 
	    // closed by interrupt exception and closes the socket
	    log.debug("Waiting for the " + THREAD_NAME + " to finish");
	    try {
		if(running) { //if it is still alive
		    acceptorThread.join();
		}
	    } catch (InterruptedException e) {
		// No - op: thread finished :)
	    }
	    log.debug("Acceptor thread stopped... cleaning up");


	    beforeStop(); //maybe it's redundant 

	}

	//listeners are notified when the acceptor Thread lunches the interrupted exception in the mothod run()

    }

    /* (non-Javadoc)
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {

	try {
	    synchronized (statusLock) {
		while(!(incomingSocket != null && incomingSocket.isOpen() && incomingSocket.socket().isBound())) {
		    log.trace("Rebinding sockets");
		    incomingSocket = ServerSocketChannel.open();
		    incomingSocket.socket().bind(address);
		}
	    }
	    running = true;
	    synchronized(this) {
		notifyAll(); //notifies the thread that started the acceptor that the acceptor has started.
	    }
	    listenForConnections();
	} catch (ClosedByInterruptException e) {
	    log.warn("The " + THREAD_NAME + " thread was closed by an external interrupt");
	    beforeStop();
	    notifyStop(e);
	    return;
	} catch(ClosedChannelException e){
	    log.error("Error: ServerSocketChannel closed: " +e,e);
	    beforeStop();
	    notifyStop(e);
	    return;
	} catch(IOException e) {
	    log.error("Error: Unexpetcer IO error: " + e, e);
	    beforeStop();
	    notifyStop(e);
	    return;
	} catch(Throwable e) {
	    log.error("FATAL: unexpexted error: " +e ,e );
	    beforeStop();
	    notifyStop(e);
	    return;
	}
    }

    /**
     * Notifies every listener that the dispatcher has stopped.
     * This method is called by the acceptor thread
     * @param e the cause of the stop
     */
    private void notifyStop(Throwable e) {
	synchronized (registeredListeners) {
	    for(AcceptorListener l : registeredListeners) {
		l.acceptorStopped(e);
	    }
	}

    }


    /**
     * Cleans status variables of this acceptor instance. Should be called after
     * server shutdown or unexpected fatal errors.
     */
    //CONCURRENCY: It won't be of any harm not to synchronise this method since running is volatile,
    //			   and the rest of the operations are idempotent.
    private void beforeStop() {
	running = false;

	try {
	    synchronized (statusLock) {


		incomingSocket.close();
		incomingSocket = null;
	    }
	} catch (Exception e) {
	    //don't care, I'm quitting.
	}

    }

    /**
     * Implements the "listening-cycle"
     */
    // This method should be run only from the acceptor thread, that should be one and only one for
    // acceptor instance. Anyway I need to synchronise accesses to other state variables from other
    // threads
    private void listenForConnections() throws IOException {
	while(running) {
	    SocketChannel connection = incomingSocket.accept();
	    if (connection == null) {
		continue;
	    }
	    ProtocolHandler handler = null; //the handler for data coming in from the newly accepted connection

	    /*
	     * No one can change the dispatcher or the class for protocol handlers 
	     * while the connection is being accepted and set-up on the dispatcher
	     */
	    synchronized (statusLock) {
		try {
		    handler = protocolHandlerClass.newInstance();
		} catch (Exception e) {
		    String msg ="Unable to instantiate protocol handler: " +e;
		    log.warn(msg,e);
		}

		dispatcher.register(connection, handler);
	    }
	}
    }

    /**
     * Returns the thread that is currently running the acceptor
     * @return the thread that is currently running the acceptor
     */
    public synchronized Thread getAcceptorThread(){
	return this.acceptorThread;
    }

    /* ************************************* */
    /* Listener Registration related methods */
    /* ************************************* */

    @Override
    //CONCURRENCY: synchronised on the set itself
    public boolean registerListener(AcceptorListener listener) {
	synchronized (registeredListeners) {
	    return registeredListeners.add(listener);
	}
    }

    @Override
    //CONCURRENCY: synchronised on the set itself
    public boolean removeListener(AcceptorListener listener) {
	synchronized (registeredListeners) {
	    return registeredListeners.remove(listener);
	}
    }



}
